package org.xuk.spring.ws.service.impl;

import com.google.gson.Gson;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;
import org.xuk.spring.ws.constants.WsConstants;
import org.xuk.spring.ws.model.BoardCastMessage;
import org.xuk.spring.ws.model.SingleMessage;
import org.xuk.spring.ws.service.ProducerClient;

import java.nio.charset.StandardCharsets;
import java.util.UUID;

@Slf4j
@Component
public class RabbitMqProducerClient implements ProducerClient, RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnsCallback {

    private final RabbitTemplate rabbitTemplate;

    private final Gson gson;

    public RabbitMqProducerClient(RabbitTemplate rabbitTemplate, Gson gson) {
        this.rabbitTemplate = rabbitTemplate;
        this.gson = gson;
    }

    @Override
    public void singleSend(SingleMessage singleMessage) {
        Message message = MessageBuilder.withBody(gson.toJson(singleMessage).getBytes(StandardCharsets.UTF_8))
                .build();
        Message message1 = setMessageDefault(message, generateId());
        sendMessage(WsConstants.SINGLE_EXCHANGE, message1);
    }

    @Override
    public void boardCastSend(BoardCastMessage boardCastMessage) {
        Message message = MessageBuilder.withBody(gson.toJson(boardCastMessage).getBytes(StandardCharsets.UTF_8))
                .build();
        Message message1 = setMessageDefault(message, generateId());
        sendMessage(WsConstants.BOARD_CAST_EXCHANGE, message1);
    }

    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        if (correlationData == null) {
            return;
        }
        String messageId = correlationData.getId();
        ReturnedMessage returned = correlationData.getReturned();
        if (returned != null) {
            log.info("消息被退回,messageId={}", messageId);
        }
        if (ack) {
            log.debug("消息发布成功,messageId={}", messageId);
        } else {
            log.error("消息发布失败,messageId={},cause={}", messageId, cause);
        }
    }

    private void sendMessage(String exchange, Message message) {
        rabbitTemplate.setConfirmCallback(this);
        rabbitTemplate.setReturnCallback(this);
        CorrelationData correlationData = new CorrelationData(message.getMessageProperties().getMessageId());
        rabbitTemplate.convertAndSend(exchange, WsConstants.ROUTING_KEY, message, correlationData);

    }

    private Message setMessageDefault(Message message, String messageId) {
        MessageProperties messageProperties = message.getMessageProperties();
        if (messageProperties == null) {
            messageProperties = new MessageProperties();
        }
        messageProperties.setMessageId(messageId);
        return MessageBuilder.withBody(message.getBody()).andProperties(messageProperties).build();
    }

    private String generateId() {
        return UUID.randomUUID().toString().replace("-", "");
    }

    @Override
    public void returnedMessage(ReturnedMessage returned) {
        log.error("消息被退回，messageId={},replyCode,replyText={},exchange={},routingKey={}",
                returned.getMessage().getMessageProperties().getMessageId(),
                returned.getReplyCode(), returned.getReplyText(), returned.getRoutingKey());
    }
}
